package com.luych.toolbox.data.service.job;

import com.luych.toolbox.common.enums.DataSourceStatus;
import com.luych.toolbox.common.event.base.service.publisher.WebSocketEventPublisher;
import com.luych.toolbox.common.event.base.service.publisher.WebSocketLetterPublisher;
import com.luych.toolbox.common.event.base.service.publisher.WebSocketNoticePublisher;
import com.luych.toolbox.common.feign.param.gateway.center.WebSocketEvent;
import com.luych.toolbox.data.service.constant.Constant;
import com.luych.toolbox.data.service.core.CoreBuilder;
import com.luych.toolbox.data.service.core.CoreConnection;
import com.luych.toolbox.data.service.dao.DataSourceDao;
import com.luych.toolbox.data.service.entity.DataSource;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.bus.BusProperties;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.scheduling.quartz.QuartzJobBean;

public class DataSourceHealthCheckJob extends QuartzJobBean {
    private static final Logger LOGGER = LoggerFactory.getLogger(DataSourceHealthCheckJob.class);

    private final BusProperties busProperties;
    private final ApplicationEventPublisher applicationEventPublisher;
    private final DataSourceDao dataSourceDao;

    public DataSourceHealthCheckJob(BusProperties busProperties,
                                    ApplicationEventPublisher applicationEventPublisher,
                                    DataSourceDao dataSourceDao) {
        this.busProperties = busProperties;
        this.applicationEventPublisher = applicationEventPublisher;
        this.dataSourceDao = dataSourceDao;
    }

    @Override
    protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
        String id = context.getJobDetail().getJobDataMap().getString("id");

        WebSocketLetterPublisher letterPublisher = new WebSocketLetterPublisher(busProperties, applicationEventPublisher);
        WebSocketNoticePublisher noticePublisher = new WebSocketNoticePublisher(busProperties, applicationEventPublisher);
        WebSocketEventPublisher eventPublisher = new WebSocketEventPublisher(busProperties, applicationEventPublisher);

        DataSource dataSource = dataSourceDao.findById(id).orElseThrow(() ->
                new JobExecutionException(Constant.DATA_SOURCE_NOT_FOUND));

        if (DataSourceStatus.BACKUPING.equals(dataSource.getStatus()) || DataSourceStatus.REVOCERING.equals(dataSource.getStatus())) {
            LOGGER.info("the data source [{}] is backuping or recovering, ignore this healthy check", dataSource.getUrl());
            return;
        }

        CoreConnection connection = CoreBuilder.buildConnection(dataSource);
        DataSourceStatus status;
        LOGGER.info("start to connect data source {}...", dataSource.getUrl());
        try {
            connection.ping();
            status = DataSourceStatus.NORMAL;
            LOGGER.info("data source {} connect success", dataSource.getUrl());
        } catch (Exception e) {
            LOGGER.error("data source {} connect error: {}", dataSource.getUrl(), e.getMessage());
            status = DataSourceStatus.ABNORMAL;
        }
        if (!dataSource.getStatus().equals(status)) {
            String message = "数据源连接状态通知";
            if (dataSource.getStatus() != DataSourceStatus.ABNORMAL && status == DataSourceStatus.ABNORMAL) {
                String content = "数据源[" + dataSource.getName() + "]连接异常，请注意修复！";
                letterPublisher.error(dataSource.getOwnerId(), message, content);
                noticePublisher.error(dataSource.getOwnerId(), content);
                eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_SOURCE_ABNORMAL, dataSource.getId()));
            } else if (dataSource.getStatus() != DataSourceStatus.NORMAL && status == DataSourceStatus.NORMAL) {
                String content = "数据源[" + dataSource.getName() + "]连接正常。";
                letterPublisher.success(dataSource.getOwnerId(), message, content);
                noticePublisher.success(dataSource.getOwnerId(), content);
                eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_SOURCE_NORMAL, dataSource.getId()));
            }

            dataSource.setStatus(status);
            dataSourceDao.save(dataSource);
        }
    }
}
